package com.lagou.filter;

import com.lagou.bean.RpcInfo;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter, Runnable{
    Map<String, List<RpcInfo>> rpcInfoMap = new ConcurrentHashMap<>();
    public TPMonitorFilter(){
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(this, 60, 5, TimeUnit.SECONDS);
    }
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        Result result = null;
        long begin = System.currentTimeMillis();
        try {
            result = invoker.invoke(invocation);
        } catch (RpcException e) {
            System.out.println(e.getMessage());
        }
        long end = System.currentTimeMillis();

        addRpcInfo(new RpcInfo(invocation.getMethodName(), end - begin, end));

        return result;
    }

    private void addRpcInfo(RpcInfo rpcInfo){
        List<RpcInfo> rpcInfos = this.rpcInfoMap.get(rpcInfo.getMethodName());
        if(rpcInfos == null){
            rpcInfos = new ArrayList<>();
            rpcInfoMap.put(rpcInfo.getMethodName(), rpcInfos);
        }
        rpcInfos.add(rpcInfo);
    }

    @Override
    public void run() {
        printTP(90);
        printTP(99);
    }

    private void printTP(int tp) {
        rpcInfoMap.values().parallelStream().forEach(rpcInfos -> {
            long currentTimeMillis = System.currentTimeMillis();
            rpcInfos = rpcInfos.stream().filter(rpcInfo -> currentTimeMillis - rpcInfo.getCallTime() < 60000).collect(Collectors.toList());
            rpcInfos.sort(new Comparator<RpcInfo>() {
                @Override
                public int compare(RpcInfo o1, RpcInfo o2) {
                    return (int) (o1.getSpendTime() - o2.getSpendTime());
                }
            });
            RpcInfo rpcInfo = rpcInfos.get(rpcInfos.size()  * tp / 100 - 1);
            System.out.println(rpcInfo.getMethodName() + " TP" + tp + ":" + rpcInfo.getSpendTime());
        });
    }
}
